package com.udf.flink.scala.apitest.evictor


import java.text.SimpleDateFormat
import java.time.Duration

import com.udf.flink.scala.apitest.checkpoint.Obj1
import com.udf.flink.scala.apitest.utils.taobao.UserBehavior
import com.udf.flink.scala.apitest.window.{MinDataReduceFunction, MyWaterMarker, PrintWindElenemt, PrintWindElenemtApply}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object TestCountEvictor extends App {
  val environment:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

  val stream1: DataStream[String] = environment.socketTextStream("localhost",9999)
  environment.setParallelism(1)
  val stream2: DataStream[Obj1] = stream1.map(data => {
    val arr = data.split(",")
    Obj1(arr(0), arr(1), arr(2).toInt)
  })

  // 设置一个窗口时间是 10 秒的窗口
  stream2
    .assignTimestampsAndWatermarks(
      WatermarkStrategy
//        .forMonotonousTimestamps()
          .forBoundedOutOfOrderness(Duration.ofSeconds(10L))
        .withTimestampAssigner(new  MyWaterMarker)
    )
    .keyBy(_.id)
   .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    //TumblingEventTimeWindows TumblingProcessingTimeWindows 生成的
    // 使用 CountEvictor 移除器,设置在开窗之前移除数据 只保留10个数据
    .evictor(CountEvictor.of(3L,false))
    .apply(new PrintWindElenemtApply)
//    .reduce(new MinDataReduceFunction)
//    .process(new PrintWindElenemt)
    .print()
  environment.execute()
}
/*
Evictor可在Window Function执行前或后，从原Window中剔除元素。

本文总结Flink DataStream Window内置的三种剔除器: CountEvictor、DeltaEvictor、TimeEvictor的剔除原理及使用。

CountEvictor: 数量剔除器。在Window中保留指定数量的元素，并从窗口头部开始丢弃其余元素。

DeltaEvictor: 阈值剔除器。计算Window中最后一个元素与其余每个元素之间的增量，丢弃增量大于或等于阈值的元素。

TimeEvictor: 时间剔除器。保留Window中最近一段时间内的元素，并丢弃其余元素。

注意: 因为在Window Function执行前剔除比较好理解，所以这里的示例均为在Window Function之后剔除元素。

原文链接：https://blog.csdn.net/wangpei1949/article/details/102996170
 */